/*
 * Copyright (C) 2013, 2014 Brett Wooldridge
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.zaxxer.hikari.util;

import com.zaxxer.hikari.util.ConcurrentBag.IConcurrentBagEntry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.ref.WeakReference;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static com.zaxxer.hikari.util.ClockSource.currentTime;
import static com.zaxxer.hikari.util.ClockSource.elapsedNanos;
import static com.zaxxer.hikari.util.ConcurrentBag.IConcurrentBagEntry.*;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.locks.LockSupport.parkNanos;

/**
 * This is a specialized concurrent bag that achieves superior performance
 * to LinkedBlockingQueue and LinkedTransferQueue for the purposes of a
 * connection pool.  It uses ThreadLocal storage when possible to avoid
 * locks, but resorts to scanning a common collection if there are no
 * available items in the ThreadLocal list.  Not-in-use items in the
 * ThreadLocal lists can be "stolen" when the borrowing thread has none
 * of its own.  It is a "lock-less" implementation using a specialized
 * AbstractQueuedLongSynchronizer to manage cross-thread signaling.
 *
 * Note that items that are "borrowed" from the bag are not actually
 * removed from any collection, so garbage collection will not occur
 * even if the reference is abandoned.  Thus care must be taken to
 * "requite" borrowed objects otherwise a memory leak will result.  Only
 * the "remove" method can completely remove an object from the bag.
 *
 * @author Brett Wooldridge
 *
 * @param <T> the templated type to store in the bag
 */
public class ConcurrentBag<T extends IConcurrentBagEntry> implements AutoCloseable
{
   private static final Logger LOGGER = LoggerFactory.getLogger(ConcurrentBag.class);

   /** 保存所有的资源数据 */
   private final CopyOnWriteArrayList<T> sharedList;
   /** 是否启动弱引用 */
   private final boolean weakThreadLocals;
   /** 当前线程中保存连接器资源 */
   private final ThreadLocal<List<Object>> threadList;
   /** 监听器，这里 HikariPool 实现了这个接口 */
   private final IBagStateListener listener;
   /** 当前等待获取连接器的数量 */
   private final AtomicInteger waiters;
   /** 判断当前线程池是否关闭 */
   private volatile boolean closed;
   /** 交接队列，线程池会将连接器对象存储到当前队列中 */
   private final SynchronousQueue<T> handoffQueue;

   /**
    * 线程池中包装对象的顶级接口
    * 创建人: zhonghaijun
    * 创建时间: 2023-08-03 09:31:37
    */
   public interface IConcurrentBagEntry
   {
      //没有被使用
      int STATE_NOT_IN_USE = 0;
      //被使用
      int STATE_IN_USE = 1;
      //移除
      int STATE_REMOVED = -1;
      //保留
      int STATE_RESERVED = -2;


      /**
       *
       *
       * @param expectState 期望状态
       * @param newState    新状态
       * @return boolean
       * @author zhonghaijun
       * @date 2023-08-03 09:32:38
       */
      boolean compareAndSet(int expectState, int newState);
      void setState(int newState);
      int getState();
   }

   public interface IBagStateListener
   {
      void addBagItem(int waiting);
   }

   /**
    * 构造方法传入一个监听器，HikariPool 实现了这个监听器接口
    *
    * @param listener the IBagStateListener to attach to this bag
    */
   public ConcurrentBag(final IBagStateListener listener)
   {
      this.listener = listener;
      //是否使用弱引用，通过 com.zaxxer.hikari.useWeakReferences 配置进行设置
      this.weakThreadLocals = useWeakThreadLocals();
      //创建 SynchronousQueue 交接队列，指定使用公平锁的策略，FIFO先进先出的原则，否则就无序进行获取
      this.handoffQueue = new SynchronousQueue<>(true);
      //初始化等待获取连接器的数量
      this.waiters = new AtomicInteger();
      //设置list，存放所有的资源
      this.sharedList = new CopyOnWriteArrayList<>();
      //是否使用弱引用，如果开启了弱引用直接采用 ArrayList，否则采用 FastList进行数据的存储
      if (weakThreadLocals) {
         this.threadList = ThreadLocal.withInitial(() -> new ArrayList<>(16));
      }
      else {
         this.threadList = ThreadLocal.withInitial(() -> new FastList<>(IConcurrentBagEntry.class, 16));
      }
   }

   /**
    * The method will borrow a BagEntry from the bag, blocking for the
    * specified timeout if none are available.
    *
    * @param timeout how long to wait before giving up, in units of unit
    * @param timeUnit a <code>TimeUnit</code> determining how to interpret the timeout parameter
    * @return a borrowed instance from the bag or null if a timeout occurs
    * @throws InterruptedException if interrupted while waiting
    */
   public T borrow(long timeout, final TimeUnit timeUnit) throws InterruptedException
   {
      // 优先从当前线程中获取连接器资源
      final List<Object> list = threadList.get();
      //从尾部开始遍历资源列表
      for (int i = list.size() - 1; i >= 0; i--) {
         //从当前线程中移除资源
         final Object entry = list.remove(i);
         @SuppressWarnings("unchecked")
         final T bagEntry = weakThreadLocals ? ((WeakReference<T>) entry).get() : (T) entry;
         //判断当前连接对象状态是否可以用，元素可能被其它线程偷取
         if (bagEntry != null && bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
            return bagEntry;
         }
      }

      // 将等待的数量加1
      final int waiting = waiters.incrementAndGet();
      try {
         //从共享资源列表中获取实例对象
         for (T bagEntry : sharedList) {
            //通过CAS修改当前实例资源的状态，判断是否修改成功
            if (bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
               // 如果大于1，那么说明不止一个线程再等待获取资源（可能偷取了别人的元素），需要通知连接池，需要创建一个连接实例对象
               if (waiting > 1) {
                  listener.addBagItem(waiting - 1);
               }
               //返回连接资源
               return bagEntry;
            }
         }
         //通知连接池，现在需要创建一个连接实例对象，这里listener是 HikariPool
         listener.addBagItem(waiting);
         //设置超时的时间
         timeout = timeUnit.toNanos(timeout);
         do {
            final long start = currentTime();
            //从交接队列中进行数据的获取，
            final T bagEntry = handoffQueue.poll(timeout, NANOSECONDS);
            //如果获取已经超时了或者修改状态成功直接返回
            if (bagEntry == null || bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
               return bagEntry;
            }
            //减去当前消耗的时间
            timeout -= elapsedNanos(start);
         } while (timeout > 10_000);

         return null;
      }
      finally {
         //等待的数量减1
         waiters.decrementAndGet();
      }
   }

   /**
    * 归还连接资源
    *
    * @param bagEntry the value to return to the bag
    * @throws NullPointerException if value is null
    * @throws IllegalStateException if the bagEntry was not borrowed from the bag
    */
   public void requite(final T bagEntry)
   {
      //先将资源的状态设置为未使用
      bagEntry.setState(STATE_NOT_IN_USE);
      //判断当前是否还有线程在等待资源的获取，如果有正在等待的线程，那么直接将资源放入交接队列中
      for (int i = 0; waiters.get() > 0; i++) {
         //如果当前资源为未使用状态，那么直接放入交接队列中；如果是其它的状态直接返回
         if (bagEntry.getState() != STATE_NOT_IN_USE || handoffQueue.offer(bagEntry)) {
            return;
         }
         else if ((i & 0xff) == 0xff) { //如果当前遍历的数为255还有线程在等待获取，那么就休眠10微秒
            parkNanos(MICROSECONDS.toNanos(10));
         }
         else {
            //否则就通知cpu让出当前线程的执行权，等其它的高优先级的线程执行完毕后再执行
            Thread.yield();
         }
      }
      //如果没有正在获取的线程，那么直接将资源放入当前线程的资源列表中
      final List<Object> threadLocalList = threadList.get();
      //如果当前线程的资源列表超过了50个了，就不需要当前实例资源了，等待销毁即可
      if (threadLocalList.size() < 50) {
         //将资源添加到本地线程的资源列表中
         threadLocalList.add(weakThreadLocals ? new WeakReference<>(bagEntry) : bagEntry);
      }
   }

   /**
    * Add a new object to the bag for others to borrow.
    *
    * @param bagEntry an object to add to the bag
    */
   public void add(final T bagEntry)
   {
      //判断连接池是否已经关闭
      if (closed) {
         LOGGER.info("ConcurrentBag has been closed, ignoring add()");
         throw new IllegalStateException("ConcurrentBag has been closed, ignoring add()");
      }
      //直接将连接器资源对象放入到共享的资源列表中
      sharedList.add(bagEntry);
      // 判断当前是否有等待连接资源的线程，判断当前创建的连接资源是否可以被使用，然后将资源方式交接队列中，如果失败了，那么就需要进行自旋操作
      while (waiters.get() > 0 && bagEntry.getState() == STATE_NOT_IN_USE && !handoffQueue.offer(bagEntry)) {
         Thread.yield();
      }
   }

   /**
    * Remove a value from the bag.  This method should only be called
    * with objects obtained by <code>borrow(long, TimeUnit)</code> or <code>reserve(T)</code>
    *
    * @param bagEntry the value to remove
    * @return true if the entry was removed, false otherwise
    * @throws IllegalStateException if an attempt is made to remove an object
    *         from the bag that was not borrowed or reserved first
    */
   public boolean remove(final T bagEntry)
   {
      //判断当前资源的状态是否为使用中或者预留中
      if (!bagEntry.compareAndSet(STATE_IN_USE, STATE_REMOVED) && !bagEntry.compareAndSet(STATE_RESERVED, STATE_REMOVED) && !closed) {
         LOGGER.warn("Attempt to remove an object from the bag that was not borrowed or reserved: {}", bagEntry);
         return false;
      }
      //将资源从共享列表中移除
      final boolean removed = sharedList.remove(bagEntry);
      if (!removed && !closed) {
         LOGGER.warn("Attempt to remove an object from the bag that does not exist: {}", bagEntry);
      }
      //从本地资源表中移除
      threadList.get().remove(bagEntry);

      return removed;
   }

   /**
    * Close the bag to further adds.
    */
   @Override
   public void close()
   {
      closed = true;
   }

   /**
    * This method provides a "snapshot" in time of the BagEntry
    * items in the bag in the specified state.  It does not "lock"
    * or reserve items in any way.  Call <code>reserve(T)</code>
    * on items in list before performing any action on them.
    *
    * @param state one of the {@link IConcurrentBagEntry} states
    * @return a possibly empty list of objects having the state specified
    */
   public List<T> values(final int state)
   {
      //获取到某个状态的资源列表
      final List<T> list = sharedList.stream().filter(e -> e.getState() == state).collect(Collectors.toList());
      Collections.reverse(list);
      return list;
   }

   /**
    * This method provides a "snapshot" in time of the bag items.  It
    * does not "lock" or reserve items in any way.  Call <code>reserve(T)</code>
    * on items in the list, or understand the concurrency implications of
    * modifying items, before performing any action on them.
    *
    * @return a possibly empty list of (all) bag items
    */
   @SuppressWarnings("unchecked")
   public List<T> values()
   {
      //克隆一次列表
      return (List<T>) sharedList.clone();
   }

   /**
    * The method is used to make an item in the bag "unavailable" for
    * borrowing.  It is primarily used when wanting to operate on items
    * returned by the <code>values(int)</code> method.  Items that are
    * reserved can be removed from the bag via <code>remove(T)</code>
    * without the need to unreserve them.  Items that are not removed
    * from the bag can be make available for borrowing again by calling
    * the <code>unreserve(T)</code> method.
    *
    * @param bagEntry the item to reserve
    * @return true if the item was able to be reserved, false otherwise
    */
   public boolean reserve(final T bagEntry)
   {
      //将当前资源的状态设置为保留状态让其线程不能借用，可以做一些逻辑操作
      return bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_RESERVED);
   }

   /**
    * This method is used to make an item reserved via <code>reserve(T)</code>
    * available again for borrowing.
    *
    * @param bagEntry the item to unreserve
    */
   @SuppressWarnings("SpellCheckingInspection")
   public void unreserve(final T bagEntry)
   {
      //将当前资源的状态设置为未使用状态，让其线程可以借用
      if (bagEntry.compareAndSet(STATE_RESERVED, STATE_NOT_IN_USE)) {
         //在判断是否有线程在等待资源的获取
         while (waiters.get() > 0 && !handoffQueue.offer(bagEntry)) {
            Thread.yield();
         }
      }
      else {
         LOGGER.warn("Attempt to relinquish an object to the bag that was not reserved: {}", bagEntry);
      }
   }

   /**
    * Get the number of threads pending (waiting) for an item from the
    * bag to become available.
    *
    * @return the number of threads waiting for items from the bag
    */
   public int getWaitingThreadCount()
   {
      return waiters.get();
   }

   /**
    * Get a count of the number of items in the specified state at the time of this call.
    *
    * @param state the state of the items to count
    * @return a count of how many items in the bag are in the specified state
    */
   public int getCount(final int state)
   {
      int count = 0;
      for (IConcurrentBagEntry e : sharedList) {
         if (e.getState() == state) {
            count++;
         }
      }
      return count;
   }

   public int[] getStateCounts()
   {
      final int[] states = new int[6];
      for (IConcurrentBagEntry e : sharedList) {
         ++states[e.getState()];
      }
      states[4] = sharedList.size();
      states[5] = waiters.get();

      return states;
   }

   /**
    * Get the total number of items in the bag.
    *
    * @return the number of items in the bag
    */
   public int size()
   {
      return sharedList.size();
   }

   public void dumpState()
   {
      sharedList.forEach(entry -> LOGGER.info(entry.toString()));
   }

   /**
    * Determine whether to use WeakReferences based on whether there is a
    * custom ClassLoader implementation sitting between this class and the
    * System ClassLoader.
    *
    * @return true if we should use WeakReferences in our ThreadLocals, false otherwise
    */
   private boolean useWeakThreadLocals()
   {
      try {
         if (System.getProperty("com.zaxxer.hikari.useWeakReferences") != null) {   // undocumented manual override of WeakReference behavior
            return Boolean.getBoolean("com.zaxxer.hikari.useWeakReferences");
         }

         return getClass().getClassLoader() != ClassLoader.getSystemClassLoader();
      }
      catch (SecurityException se) {
         return true;
      }
   }
}
